/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.ybk.e2p.yb.tools;

import cn.ac.iie.di.ban.ycl.ybk.e2p.yb.main.BanBkE2p;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import java.io.FileInputStream;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import org.apache.log4j.Logger;

/**
 *
 * @author wmr
 */
public class RocketmqUtil {

    public String status = "";
    public String desc = "";
    private static Logger logger = Logger.getLogger(RocketmqUtil.class);
//    static String filePath = RocketmqUtil.class.getResource("/").getFile() + "rocketmq.properties";
    static String filePath = BanBkE2p.confDir + "/rocketmq.properties";
//    public boolean produceData(String filePath, String subjobid, Map<String, String> rsMap) throws MQClientException, InterruptedException {
//    public static void main(String[] args) throws MQClientException, InterruptedException {

    public static boolean produceData(String filePath) throws MQClientException, InterruptedException {
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
        logger.info(" RocketmqUtil start " + df.format(new Date()));
        boolean flag = true;
        String name_server_list = null;
        String topic = null;
        int send_times;
        Properties prop = new Properties();
        try {
            InputStream in = new FileInputStream("rocketmq.properties");
            prop.load(in);
            in.close();
            name_server_list = prop.getProperty("name_server_list").trim();
            topic = prop.getProperty("topic").trim();
            send_times = Integer.parseInt(prop.getProperty("send_times").trim());//send_times
        } catch (Exception e) {
            flag = false;
            logger.error(" rocketmq.properties file read fail! " + e.getMessage(), e);
            return flag;
        }

        DefaultMQProducer producer = new DefaultMQProducer("11");
//        producer.setNamesrvAddr("192.168.11.95:9876");
        producer.setNamesrvAddr(name_server_list);
        producer.setInstanceName("111");
//        producer.setMaxMessageSize(8 * 1024 * 1024);
//        producer.setCreateTopicKey(topic);

        try {
            producer.start();
            logger.info(" RocketmqUtil init producer successfully!");
        } catch (Exception ex) {
            logger.error("init producer unsuccessfully for " + ex.getMessage(), ex);
            flag = false;
            return flag;
        }

        for (int i = 0; i < send_times; i++) {
            try {
                Message msg = new Message(topic, (filePath).getBytes());
//                Message msg = new Message("2a4_test", (filePath).getBytes());
                SendResult sendResult = producer.send(msg);
                logger.info("SEND SUCCESS --------" + sendResult);
                if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
                    flag = true;
                    logger.info(" producer send sucess " + filePath);
                } else {
                    flag = false;
                    logger.error(" producer send error ");
                }
            } catch (Exception ex) {
                flag = false;
                logger.error(" producer send Exception " + ex.getMessage(), ex);
            }

        }

        producer.shutdown();
        logger.info(" RocketmqUtil end " + df.format(new Date()));
        return flag;
    }

    public static void main(String[] args) throws MQClientException, InterruptedException {
        RocketmqUtil.produceData("d://520.txt");
    }
}
